const mqtt = require('mqtt');

class MqttConnector {
  constructor(brokerUrl, clientId) {
    this.brokerUrl = brokerUrl;
    this.clientId = clientId || `client-${Math.random().toString(16).substr(2, 8)}`;
    this.client = null;
    this.messageListeners = [];
  }

  connect() {
    return new Promise((resolve, reject) => {
      this.client = mqtt.connect(this.brokerUrl, {
        clientId: this.clientId,
        clean: true,
        connectTimeout: 4000,
        reconnectPeriod: 1000
      });

      this.client.on('connect', () => {
        console.log(`Connected to MQTT broker: ${this.brokerUrl}`);
        resolve(this.client);
      });

      this.client.on('error', (error) => {
        console.error(`MQTT connection error: ${error.message}`);
        reject(error);
      });
    });
  }

  subscribe(topic, qos = 0) {
    return new Promise((resolve, reject) => {
      if (!this.client) {
        reject(new Error('Not connected to MQTT broker'));
        return;
      }

      this.client.subscribe(topic, { qos }, (error) => {
        if (error) {
          console.error(`Subscribe error: ${error.message}`);
          reject(error);
        } else {
          console.log(`Subscribed to topic: ${topic}`);
          resolve();
        }
      });
    });
  }

  publish(topic, message, qos = 0) {
    return new Promise((resolve, reject) => {
      if (!this.client) {
        reject(new Error('Not connected to MQTT broker'));
        return;
      }

      // 如果消息是对象，自动序列化为JSON
      const messageToSend = typeof message === 'object' ? JSON.stringify(message) : message;

      this.client.publish(topic, messageToSend, { qos }, (error) => {
        if (error) {
          console.error(`Publish error: ${error.message}`);
          reject(error);
        } else {
          console.log(`Published to topic: ${topic}, message: ${messageToSend}`);
          resolve();
        }
      });
    });
  }

  onMessage(callback) {
    if (!this.client) {
      throw new Error('Not connected to MQTT broker');
    }

    this.messageListeners.push(callback);

    // 只注册一次message事件处理器
    if (this.messageListeners.length === 1) {
      this.client.on('message', (topic, message, packet) => {
        let parsedMessage = message.toString();
        try {
          // 尝试解析为JSON
          parsedMessage = JSON.parse(parsedMessage);
        } catch (error) {
          // 解析失败，记录错误并保持原始字符串
          console.error(`MQTT消息解析失败: ${error.message}, 主题: ${topic}`);
        }

        // 通知所有监听器
        this.messageListeners.forEach(listener => {
          try {
            listener(topic, parsedMessage, packet);
          } catch (listenerError) {
            console.error(`消息监听器错误: ${listenerError.message}`);
          }
        });
      });
    }
  }

  disconnect() {
    if (this.client) {
      this.client.end();
      console.log('Disconnected from MQTT broker');
    }
  }
}

module.exports = MqttConnector;